//
// Created by 成飞 on 2024/3/5.
//
#include "leveldb.h"
#include "leveldb/write_batch.h"
#include "leveldb/db.h"
#include "config/default_conf_info.h"
#include "RecordVal.h"
#include <iostream>
#include <cassert>
#include <string>
#include <filesystem>
#include <fcntl.h>
#include <unistd.h>
#include "util/thread_pool.hpp"
#include "config/config_data.h"

namespace lsm{

    using std::min;
    using std::max;

    LevelDB::LevelDB(std::string &lsmPath) : lsmPath(lsmPath) {
            init_db();

    }

    bool LevelDB::put(std::string_view key, std::string_view val) {
        leveldb::Slice k{key.data(), key.size()};
        leveldb::Slice v{val.data(), sizeof (lsm::LsmRecordVal)};
        auto status = db->Put(leveldb::WriteOptions(), k, v);
        return status.ok();
    }

    bool LevelDB::batch_put(const std::vector<std::pair<std::string_view, std::string_view>> &kvPairBatch) {
        leveldb::WriteBatch batch;
        size_t len = kvPairBatch.size();
        for (int i = 0; i < len; ++i) {
            leveldb::Slice key(kvPairBatch[i].first.data(), kvPairBatch[i].first.size());
            leveldb::Slice val(kvPairBatch[i].second.data(), sizeof (lsm::LsmRecordVal));
            batch.Put(key, val);
        }
        auto status = db->Write(leveldb::WriteOptions(), &batch);
        return status.ok();
    }

    bool LevelDB::query(std::string_view key, std::string& ret) {
        leveldb::Slice k{key.data(), key.size()};
        auto status = db->Get(*(pReadOption.load()), k, &ret);
        return status.ok();
    }

    bool LevelDB::range_query(const string &startKey, const string &endKey, int len, vector<pair<string , lsm::LsmRecordVal>>& valLocations, array<lsm::ScanInfo*, FILE_NO_MAX>& scanIndex) {
        valLocations.reserve(len);
        auto it = db->NewIterator(*(pReadOption.load()));
        int resultNum = 0;
        for (it->Seek(startKey); it->Valid() && resultNum < len && it->key().ToString() < endKey; it->Next(), ++resultNum) {
            auto key = it->key().ToString();
            auto val = it->value().ToString();
            auto &recordVal = *reinterpret_cast<lsm::LsmRecordVal *>(const_cast<char *>(val.data()));
            valLocations.emplace_back(it->key().ToString(), recordVal);

            auto shouldScan = scanIndex[recordVal.fileNo];
            if (shouldScan != nullptr) {
                auto &startEndPair = *shouldScan;
                startEndPair.startOff = min(startEndPair.startOff, recordVal.recordOffset);
                startEndPair.endOff = max(startEndPair.endOff, recordVal.recordOffset + (meta::f_off)recordVal.recordLen);
            }else{
                scanIndex[recordVal.fileNo] = new lsm::ScanInfo{recordVal.recordOffset,recordVal.recordOffset+recordVal.recordLen};
            }
        }

        auto stat = it->status().ok();
        delete it;
        return stat;
    }

    std::mutex rqm;
    bool LevelDB::range_query2(const std::string &startKey, const std::string &endKey, int len, std::vector<std::pair<std::string , std::string>>& ret) {
        ret.reserve(len);
        auto it = db->NewIterator(*(pReadOption.load()));
        int resultNum = 0;
        static ThreadPool threadPool(conf::data.db.scan_thread_num);
        std::vector<std::shared_future<void>> manymany;
        for (it->Seek(startKey); it->Valid() && resultNum < len && it->key().ToString() < endKey; it->Next(), ++resultNum) {
            //预读优化
            std::string val = it->value().ToString(); // todo 有必要复制吗
            auto &recordVal = *reinterpret_cast<lsm::LsmRecordVal *>(const_cast<char *>(val.data()));
            auto filePathStr =  meta::fileControlInfo.find(recordVal.fileNo)->second.addr;
            auto off = recordVal.recordOffset;
            auto recordLen = recordVal.recordLen;
            int fd = fileno(meta::fileControlInfo[recordVal.fileNo].fd);
            //posix_fadvise(fd, off, recordLen, POSIX_FADV_SEQUENTIAL | POSIX_FADV_WILLNEED);
            std::string key = it->key().ToString();
//            auto many = threadPool.enqueue([key, fd, off, recordLen, &ret] {
                // Set the file offset
                if (lseek(fd, off, SEEK_SET) == -1) {
                    perror("Error setting file offset");
                    close(fd);
                }
                // Read data into buffer
                char *buffer = new char[recordLen];
                auto bytesRead = read(fd, buffer, recordLen);
                if (bytesRead == -1) {
                    perror("Error reading file");
                    free(buffer);
                    close(fd);
                    continue;
                }
                meta::f_off keyLen;
                std::memcpy(&keyLen, buffer, sizeof(meta::f_off));
                rqm.lock();
                ret.emplace_back(key, std::string{buffer, recordLen});
                rqm.unlock();
                delete[] buffer;
//            });
//            manymany.emplace_back(std::move(many));
        }
//        for (const auto& item: manymany) {
//            item.get();
//        }
        auto stat = it->status().ok();
        delete it;
        return stat;


        //预读优化
//            std::string str = it->value().ToString(); // todo 有必要复制吗
//            auto &recordVal = *reinterpret_cast<lsm::LsmRecordVal *>(const_cast<char *>(str.data()));
//            auto filePathStr =  meta::fileControlInfo.find(recordVal.fileNo)->second.addr;
//            auto off = recordVal.recordOffset;
//            auto recordLen = recordVal.recordLen;
//            int fd = fileno(meta::fileControlInfo[recordVal.fileNo].fd);
//            posix_fadvise(fd, off, recordLen, POSIX_FADV_SEQUENTIAL | POSIX_FADV_WILLNEED);
    }


    void LevelDB::init_db() {
        leveldb::Options startupOptions;
        startupOptions.create_if_missing = true;
        std::filesystem::create_directories(std::filesystem::path(lsmPath).parent_path());
        leveldb::Status status = leveldb::DB::Open(startupOptions, lsmPath, &db);
        assert(status.ok() || assert_msg("leveldb startup fail, path: " + lsmPath));
        pReadOption.store(&defaultReadOption);
    }

    void LevelDB::snapshot() {
        snapshotReadOption.snapshot = db->GetSnapshot();
        pReadOption.store(&snapshotReadOption);
    }

    void LevelDB::release_snapshot(){
        pReadOption.store(&defaultReadOption);
    }
}